package voronin;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class ConnectionPoolService<T> implements ConnectionPool<T> {
    private Queue<T> availableConnections = new ConcurrentLinkedQueue<T>();
    private Semaphore availableConnectionsSemaphore;

    private Queue<T> allConnections = new ConcurrentLinkedQueue<>();

    private int minCapacity;
    private int maxCapacity;
    private long keepAliveTime;
    private TimeUnit unit;
    private ConnectionSource<T> source;

    private AtomicBoolean isStarted = new AtomicBoolean(false);
    private AtomicInteger allConnectionsNumber;
    private PoolWiper poolWiperThread;
    private final Object closingMonitor = new Object();

    public ConnectionPoolService(int minCapacity, int maxCapacity, long keepAliveTime, TimeUnit unit, ConnectionSource<T> source) {
        this.minCapacity = minCapacity;
        this.maxCapacity = maxCapacity;
        this.keepAliveTime = keepAliveTime;
        this.unit = unit;
        this.source = source;
    }

    public void start() {
        if (minCapacity != maxCapacity) {
            startPoolWiper();
        }
        initializeConnections();
        availableConnectionsSemaphore = new Semaphore(maxCapacity);
        isStarted.set(true);
    }

    public void stop() {
        synchronized (closingMonitor) {
            isStarted.set(false);
            poolWiperThread.interrupt();
            try {
                poolWiperThread.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            availableConnectionsSemaphore.drainPermits();
            availableConnections.clear();
            closeAllConnections();
        }
    }

    private void startPoolWiper() {
        poolWiperThread = new PoolWiper();
        poolWiperThread.start();
        while (poolWiperThread.getState() != Thread.State.WAITING) {
        }
    }

    private void initializeConnections() {
        for (int i = 0; i < minCapacity; i++) {
            T newConnection = source.open();
            availableConnections.add(newConnection);
            allConnections.add(newConnection);
        }
        allConnectionsNumber = new AtomicInteger(minCapacity);
    }

    @Override
    public T get() throws InterruptedException {
        availableConnectionsSemaphore.acquire();
        return safeGet();
    }

    @Override
    public T tryGet() {
        if (availableConnectionsSemaphore.tryAcquire()) {
            return safeGet();
        }
        return null;
    }

    @Override
    public T tryGet(long time, TimeUnit unit) throws InterruptedException {
        if (availableConnectionsSemaphore.tryAcquire(time, unit)) {
            return safeGet();
        }
        return null;
    }

    private T safeGet() {
        synchronized (closingMonitor) {
            if (!isStarted.get()) {
                return null;
            }
            T connection = availableConnections.poll();
            if (connection == null) {
                connection = source.open();
                allConnections.add(connection);
                allConnectionsNumber.incrementAndGet();
            }
            return connection;
        }
    }

    @Override
    public void release(T object) {
        if (!isStarted.get()) {
            return;
        }
        availableConnections.add(object);
        availableConnectionsSemaphore.release();
        if (poolWiperThread != null) {
            poolWiperThread.connectionReleased();
        }
    }

    private void closeAllConnections() {
        while ( allConnectionsNumber.get() != 0) {
            source.close(allConnections.poll());
            allConnectionsNumber.decrementAndGet();
        }
    }


    public int currentlyOpened() {
        return allConnectionsNumber.get();
    }

    class PoolWiper extends Thread {
        final long millisTimeout = TimeUnit.MILLISECONDS.convert(keepAliveTime, unit);
        private final Object wiperMonitor = new Object();

        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    monitor();
                } catch (InterruptedException e) {
                    //Worker is interrupted from stop()
                } finally {
                    break;
                }
            }
        }

        private void monitor() throws InterruptedException {
            while (true) {
                synchronized (wiperMonitor) {
                    wiperMonitor.wait();
                }
                if (allConnectionsNumber.get() > minCapacity) {
                    Thread.sleep(millisTimeout);
                    synchronized (wiperMonitor) {
                        synchronized (closingMonitor) {
                            closeUnusedConnections();
                        }
                    }
                }
            }
        }

        public void connectionReleased() {
            synchronized (wiperMonitor) {
                wiperMonitor.notify();
            }
        }

        private void closeUnusedConnections() {
            T unusedConnection;
            int connectionsToCloseNumber = Math.min(availableConnectionsSemaphore.availablePermits(), maxCapacity - minCapacity) + allConnectionsNumber.get() - maxCapacity;
            for (int i = 0; i < connectionsToCloseNumber; i++) {
                unusedConnection = tryGet();
                source.close(unusedConnection);
                allConnections.remove(unusedConnection);
                allConnectionsNumber.decrementAndGet();
            }
        }
    }
}